package com.nami.logic.rocketmq.consumer;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.json.JSONUtil;
import com.nami.basic.bo.servicetrans.up.UpMsgWrap;
import com.nami.basic.constants.UpTopicConst;
import com.nami.basic.enums.DownMsgErrorCode;
import com.nami.basic.enums.protocol.DownCMDEnum;
import com.nami.basic.packet.Packet;
import com.nami.basic.packet.message.Message;
import com.nami.basic.packet.message.MsgUser;
import com.nami.logic.cache.conver.ConverCache;
import com.nami.logic.forward.TransDownMsgHandler;
import com.nami.logic.qos.QoSReceiveDaemonC2S;
import com.nami.logic.serializer.Serializer;
import com.nami.logic.service.*;
import com.nami.logic.utils.ConverUtil;
import com.nami.store.api.bo.UserWrap;
import com.nami.store.api.request.conversation.ConverItem;
import com.nami.store.api.request.message.GrpMsgRequest;
import com.nami.store.api.service.RemoteConverService;
import com.nami.store.api.service.RemoteL2CacheService;
import com.nami.store.api.service.RemoteMessageService;
import com.nami.store.api.response.conversation.ConversationDTO;
import com.nami.store.api.response.group.GroupDTO;
import com.nami.store.api.response.groupmember.GroupMemberDTO;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.dromara.common.core.enums.conver.ConverTypeEnum;
import org.dromara.common.core.utils.IdentityUtil;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

import static com.nami.logic.constants.GrpConfigConst._MAX_MEM_SIZE_FOR_WRITE_INBOX_MSG;
import static com.nami.logic.thread.ThreadPoolManager.GDISPATCHERS;


@Slf4j
@Component
@AllArgsConstructor
@RocketMQMessageListener(topic = UpTopicConst.C2G_TOPIC,
    consumerGroup = UpTopicConst.C2G_CONSUMER_GROUP,
    consumeMode = ConsumeMode.CONCURRENTLY)
public class GrpMsgConsumer extends AbstractMessage implements RocketMQListener<String> {

    private final PacketFactory packetFactory;

    private final ConverSyncHandler syncHandler;

    private final GrpMemberService grpMemberService;

    private final DownMsgSender mqSender;

    private final UserService userService;

    private final GroupService groupService;

    private final TransDownMsgHandler transDownMsgHandler;

    @DubboReference
    RemoteL2CacheService cacheService;

    @DubboReference
    RemoteMessageService messageService;

    @DubboReference
    RemoteConverService converService;

    @Override
    public void onMessage(String msg) {
        try {
            TimeInterval timer = DateUtil.timer();
            final UpMsgWrap upMsgWrap = JSONUtil.toBean(msg, UpMsgWrap.class);
            final Packet packet = upMsgWrap.getPacket();
            Message message = Serializer.prototype(packet.getSerial()).deserialize(packet.getBody(), Message.class);

            addSystemInfoToThisMsg(message);

            log.info("[c2g] 群号【{}】收到客户端【{}】发来的一条群消息. clientMsgId==>{} 消息内容 ：{}", message.getReceiverId(), upMsgWrap.getUserIp(), message.getClientMsgId(), message.getPayload());
            final String identityNo = IdentityUtil.buildIdentityNo(message.getSenderId(), upMsgWrap.getDeviceType());
            if (checkNullParams(upMsgWrap, message)) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(), message,
                    identityNo, upMsgWrap, DownMsgErrorCode.REQUEST_PARAMS_ERROR, packetFactory);
                log.warn("[c2g] 校验消息失败, 丢失核心参数：deviceType:{}, clientMsgId:{}", upMsgWrap.getDeviceType(), message.getClientMsgId());
                return;
            }

            if (StringUtils.isBlank(upMsgWrap.getBindingUserId()) || !message.getSenderId().equals(upMsgWrap.getBindingUserId())
                || StringUtils.isBlank(upMsgWrap.getBindingAppKey())) {
                log.warn("[c2g] 校验到乱发消息，或参数为null. clientMsgId={}", message.getClientMsgId());
                return;
            } // end if check userid .. 防止使用别人的userId 乱发消息

            UserWrap sender = userService.getUserById(upMsgWrap.getBindingAppKey(), upMsgWrap.getBindingUserId());
            if (sender == null) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.USER_NOT_EXIST, packetFactory);
                return;
            } // end check null if

            if (checkUserWasBan(upMsgWrap, sender, identityNo)) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.FROM_USER_BAN, packetFactory);
                return;
            } // end check ban if
//            log.info("[c2g] group前 clientMsgId={}, 消耗：{}ms", message.getClientMsgId(), timer.interval());

            GroupDTO groupInfo = groupService.getGroupFromCache(sender.getAppKey(), message.getReceiverId());
            if (groupInfo == null) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.FROM_USER_NOT_EXIST, packetFactory);
                return;
            } // end check group null if
            // 对群校验
            if (groupInfo.getBanSpeak() != null && groupInfo.getBanSpeak()) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.GROUP_BAN_SPEAK, packetFactory);
                return;
            } // end banspeak check
            if (!groupInfo.getAppKey().equals(sender.getAppKey())) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.USER_APP_NOT_MATCH, packetFactory);
                return;
            } // end appkey check
//            log.info("[c2g] mem前 clientMsgId={}, 消耗：{}ms", message.getClientMsgId(), timer.interval());

            List<GroupMemberDTO> allMembers = grpMemberService.getGrpMembersFromCache(message.getReceiverId());
            if (CollUtil.isEmpty(allMembers)) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.GROUP_MEMBER_NOT_EXIST, packetFactory);
                return;
            } // end group members null check

            List<GroupMemberDTO> memberMeList = allMembers.stream()
                .filter(item -> message.getSenderId().equals(item.getUserId())).collect(Collectors.toList());
            if (CollUtil.isEmpty(memberMeList)) {
                sendErrorResp(mqSender, DownCMDEnum.C2G.getCode(),
                    message, identityNo, upMsgWrap, DownMsgErrorCode.GROUP_MEMBER_NOT_CONTAIN_ME, packetFactory);
                return;
            } // end is member of this group check me
            GroupMemberDTO me = memberMeList.get(0);

            List<GroupMemberDTO> membersExcludeMe = allMembers.stream()
                .filter(item -> !message.getSenderId().equals(item.getUserId())).collect(Collectors.toList());

            MsgUser user = new MsgUser();
            user.setNick(me.getNickName());
            user.setAvatar(sender.getAvatar());
            message.setSenderInfo(user);

            if (StringUtils.isBlank(message.getCid())) {
                message.setCid(message.getReceiverId());
                message.setConversationType(ConverTypeEnum.C2G.getCode());
            }
//            log.info("[c2g] seq前 clientMsgId={}, 消耗：{}ms", message.getClientMsgId(), timer.interval());

            // seq()
            Long maxSeq = cacheService.incrSeq(buildSequenceDTO(sender.getAppKey(), message.getCid()));
            message.setSeq(maxSeq.intValue());

            // 消息重复校验
            if (QoSReceiveDaemonC2S.getInstance()
                .hasReceived(genUserMsgFingerChar(
                    IdentityUtil.buildIdentityNo(sender.getUserId() + "", upMsgWrap.getDeviceType()), message.getClientMsgId()))) {
                log.warn("[c2g] 已丢弃客户端【{}】id={}的重复消息：{}", sender.getNick(), message.getClientMsgId(), message.getPayload());
                return;
            }

            // msg入库
            // io 性能抖动，会出现几百ms 消耗
            storeMsg(message, sender.getAppKey(), membersExcludeMe);
            // 延申出会话服务。 或使用事件异步。
            // 将 client mq 发送消息选择器移到此处，发送mq消息时，采用hash到会话服务集群某台服务，减少缓存多份相同数据
            // client 移除掉hash 到logic策略
            // 目前logic 为有状态服务，存在一级缓存，缓存一致性还没有解决，采用 向redis获取版本进行升级更新？
            // TODO 2024-12-28

            // 另 单聊，群聊，继续拆分。 单，群聊服务，（qos以及转发功能组成的下发服务），会话服务, 用户服务
            long start = System.currentTimeMillis();
            updConvers(message, sender.getAppKey(), membersExcludeMe);
            long end = (System.currentTimeMillis() - start) / 1000;
            if (end > 200) {
                log.warn("[c2g] 处理会话服务出现了问题，{}ms", end);
            }
            // 延申出转发服务 。或使用线程池 进行异步，线程池策略 为 队列满使用当前线程
            // TODO 转发服务，统一单聊，群聊。
            GDISPATCHERS.execute(() -> {
                try {
                    transDownMsgHandler.dispatchMsg(upMsgWrap, membersExcludeMe, sender, null, message);
                } catch (Exception var2) {
                    log.error("[DISPATCHER-G] ======群聊转发器出现问题了。。。==>{}", var2);
                }
            });

            log.info("[c2g] 用户【{}】向群【{}】发送的群聊消息[{},{}]已处理完毕, use:{}ms", sender.getNick(),
                message.getReceiverId(), message.getClientMsgId(), message.getId(), timer.interval());
        } catch (Exception var3) {
            var3.printStackTrace();
            log.error("[c2g] 群消息出错：{}msg", var3.getMessage());
        }

    }

    /**
     * 是否扩展会话服务，提供会话 一级缓存 相关功能
     *
     * @param message
     * @param appKey
     * @param members 除了我其他群成员
     */
    private void updConvers(final Message message, final String appKey, final List<GroupMemberDTO> members) {
        // 刷新我自己会话
        final ConverItem converMe = ConverCache.get(ConverUtil.genLocalCacheConverKey(appKey, message));
        if (null == converMe) {
            ConversationDTO conversation = converService.getConverFromCacheOrDB(appKey, message.getSenderId(), message.getReceiverId());
            ConverItem needAdd = new ConverItem();
            needAdd.setAppKey(appKey);
            needAdd.setConverType(2);
            needAdd.setUserId(message.getSenderId());
            needAdd.setTargetId(message.getReceiverId());
            needAdd.setLatestReadMsgIndex(0);
            needAdd.setIfMute(Boolean.FALSE);
            needAdd.setIfTop(Boolean.FALSE);

            needAdd.setLatestMsgId(message.getId());
            needAdd.setLatestUnreadMsgIndex(message.getSeq());
            needAdd.setSyncTime(message.getTime());
            needAdd.setSortTime(message.getTime());
            if ( conversation != null) {
                needAdd.setIfMute(conversation.getIfMute());
                needAdd.setIfTop(conversation.getIfTop());

                needAdd.setTopUpdatedTime(conversation.getTopUpdatedTime());

                if (conversation.getLatestUnreadMsgIndex() != null && conversation.getLatestUnreadMsgIndex() > message.getSeq()) {
                    needAdd.setLatestMsgId(conversation.getLatestMsgId());
                    needAdd.setLatestUnreadMsgIndex(conversation.getLatestUnreadMsgIndex());
                    needAdd.setSyncTime(conversation.getSyncTime());
                    needAdd.setSortTime(conversation.getSortTime());
                    log.warn("[c2g] 更新我的会话出现问题，数据库数据为最新数据。");
                }

                needAdd.setLatestReadMsgId(conversation.getLatestReadMsgId());
                needAdd.setLatestReadMsgIndex(conversation.getLatestReadMsgIndex());
            }
            ConverCache.insertOrUpd(ConverUtil.genLocalCacheConverKey(appKey, message), needAdd);
            syncHandler.writeNow(ConverUtil.genLocalCacheConverKey(appKey, message.getSenderId(), message.getReceiverId()));
//            syncHandler.handler(ConverUtil.genLocalCacheConverKey(appKey, message));
            return;
        }
        ConverItem updConver = converMe.clone();
        updConver.setLatestMsgId(message.getId());
        updConver.setLatestUnreadMsgIndex(message.getSeq());
        updConver.setSyncTime(message.getTime());
        updConver.setSortTime(message.getTime());
        ConverCache.insertOrUpd(ConverUtil.genLocalCacheConverKey(appKey, message), updConver);
        syncHandler.writeNow(ConverUtil.genLocalCacheConverKey(appKey, message.getSenderId(), message.getReceiverId()));
//        syncHandler.handler(ConverUtil.genLocalCacheConverKey(appKey, message));

//        ConverItem newMe = ConverCache.get(ConverUtil.genLocalCacheConverKey(appKey, message));
//        if (members.size() <= 500) {
        // 直接刷新数据库
        // TODO 更新为 分发器时候，去更新
//            return;
//        }

        // 此处需要反复修改 处于临界状态 条件问题

        List<GroupMemberDTO> needQuery = new ArrayList<>();
        // 刷新缓存
        ConverItem refresh = null;
        for (GroupMemberDTO member : members) {
            final ConverItem converItem = ConverCache.get(ConverUtil.genLocalCacheConverKey(appKey, member.getUserId(), message.getReceiverId()));
            if (null == converItem) {
                needQuery.add(member);
                continue;
            }

            if (converItem.getLatestUnreadMsgIndex() != null
                && converItem.getLatestUnreadMsgIndex() > message.getSeq()) {
//                log.warn("[c2g] 会话更新失败，当前消息非最新消息，messageId={} seq={}, cacheMaxSeq={}", message.getId(),
//                    message.getSeq(), converItem.getLatestUnreadMsgIndex());
                continue;
            }
            refresh = new ConverItem();
            refresh.setId(converItem.getId());
            refresh.setAppKey(converItem.getAppKey());
            refresh.setUserId(converItem.getUserId());
            refresh.setTargetId(converItem.getTargetId());

            refresh.setLatestMsgId(message.getId());
            refresh.setLatestMsg(message.getPayload());
            refresh.setSyncTime(message.getTime());
            refresh.setSortTime(message.getTime());
            refresh.setLatestUnreadMsgIndex(message.getSeq());
            ConverCache.insertOrUpd(ConverUtil.genLocalCacheConverKey(appKey, member.getUserId(), message.getReceiverId()), refresh);
            syncHandler.handler(ConverUtil.genLocalCacheConverKey(appKey, member.getUserId(), message.getReceiverId()));

//            converSyncHandler.handler(newConverBrief(appKey, member.getUserId(), message.getReceiverId()));
        }

        List<String> userIds = needQuery.stream().map(GroupMemberDTO::getUserId).collect(Collectors.toList());
        if (CollUtil.isEmpty(userIds)) {
            return;
        }
        List<ConversationDTO> conversations = converService.getConversFromCacheOrDb(appKey, userIds, message.getReceiverId());
        Map<String, ConversationDTO> converMap = new HashMap<>();
        if (CollUtil.isNotEmpty(conversations)) {
            converMap = conversations.stream().collect(Collectors.toMap(ConversationDTO::getUserId, Function.identity()));
        }
        ConverItem converItem;
        for (String userId : userIds) {
            converItem = new ConverItem();
            converItem.setConverType(2);
            converItem.setAppKey(appKey);
            converItem.setUserId(userId);
            converItem.setTargetId(message.getReceiverId());

            ConversationDTO dbConver = converMap.get(userId);
            if (dbConver == null) {
                // 如果数据库没有，再次新建会话
                // 此处是新建会话，已读index 待处理
                converItem.setIfMute(Boolean.FALSE);
                converItem.setIfTop(Boolean.FALSE);

                converItem.setLatestMsgId(message.getId());
                converItem.setSyncTime(message.getTime());
                converItem.setSortTime(message.getTime());
                converItem.setLatestUnreadMsgIndex(message.getSeq());
                // qq 删除会话 会将会话中未读消息已读，并不删除服务端存储的会话，退群，客户端删除会话才会真正删除。所以不退群，不存在会话不存在，这种是极端情况。
                // 会话从客户端列表删除后，再次发来消息， 此时为客户端调用sdk将未读数置0，全部已读
//                converItem.setLatestReadMsgIndex(message.getSeq() - 1);
//                converItem.setLatestReadMsgTime(message.getTime()); // 或者为0
                ConverCache.insertOrUpd(ConverUtil.genLocalCacheConverKey(converItem.getAppKey(), converItem.getUserId(), converItem.getTargetId()), converItem);
//                converSyncHandler.handler(newConverBrief(appKey, converItem.getUserId(), converItem.getTargetId()));
                syncHandler.handler(ConverUtil.genLocalCacheConverKey(appKey, converItem.getUserId(), converItem.getTargetId()));
                continue;
            }

            converItem.setId(dbConver.getId());
            converItem.setSyncTime(message.getTime());
            converItem.setSortTime(message.getTime());

            if (dbConver.getLatestUnreadMsgIndex() != null
                && dbConver.getLatestUnreadMsgIndex() > message.getSeq()) {
//                log.warn("[c2g] 会话更新失败，当前消息非最新消息，messageId={} seq={}, cacheMaxSeq={}", message.getId(),
//                    message.getSeq(), converItem.getLatestUnreadMsgIndex());
                converItem.setLatestMsgId(dbConver.getLatestMsgId());
                converItem.setLatestUnreadMsgIndex(dbConver.getLatestUnreadMsgIndex());
                converItem.setSyncTime(dbConver.getSyncTime());
                converItem.setSortTime(dbConver.getSortTime());
            }
            // 此处找不到比较对象，只能从数据取默认的。
            converItem.setLatestReadMsgIndex(dbConver.getLatestReadMsgIndex());
            converItem.setLatestReadMsgTime(dbConver.getLatestReadMsgTime());
            converItem.setLatestReadMsgId(dbConver.getLatestReadMsgId());

            ConverCache.insertOrUpd(ConverUtil.genLocalCacheConverKey(converItem), converItem);
            syncHandler.handler(ConverUtil.genLocalCacheConverKey(converItem));
        }

    }

    private void storeMsg(final Message message, final String appKey, List<GroupMemberDTO> members) {
        // save
        GrpMsgRequest request = new GrpMsgRequest();
        request.setAppKey(appKey);

        request.setSenderId(message.getSenderId());
        request.setCid(message.getCid());

        request.setSeq(message.getSeq());

        request.setMessageId(message.getId());

        request.setPayload(message.getPayload());
        request.setPayloadType(message.getPayloadType());

        request.setType(message.getType());
        if (members != null && members.size() < _MAX_MEM_SIZE_FOR_WRITE_INBOX_MSG) {
            Set<String> userIds = members.stream().map(item -> item.getUserId()).collect(Collectors.toSet());
            // 批量写入: 在2，3k size 时候 会出现耗时飙高
            // 批量写入：id 提前写入，能够优化mysql 在插入时查询主键最大值 这个耗时。
            // 此处应为 群小于阈值的，且为在线群成员的，写入收件箱。逻辑待修改
            request.setReceiverIds(userIds);
        }

        request.setSendTime(new Date(message.getTime()));

        messageService.saveGrpMsg(request);
    }


}
